package org.fastsyncer.manager.listener.oracle;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.constant.MappingConstant;
import org.fastsyncer.common.entity.DatabaseConfig;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.connector.util.JDBCUtil;
import org.fastsyncer.manager.listener.AbstractListener;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class OracleListener extends AbstractListener {

    private static final Logger logger = LoggerFactory.getLogger(OracleListener.class);

    // 轮训读取增量数据间隔3秒
    private static final long MILLIS = 3000L;

    private String table;

    private String dictionary;

    private Connection conn = null;

    private Statement statement = null;

    private boolean isRunnable = true;

    @Override
    public void onClose() {
        // 标记线程状态为关闭
        isRunnable = false;
    }

    @Override
    public void run() {
        // 启动监听线程
        // 1、获取驱动
        MappingTask task = super.getMapping();
        if (null == task) {
            return;
        }
        Mapping mapping = task.getSourceMapping();
        // 2、检查驱动连接器类型
        String connector = mapping.getConnector();
        if (!StringUtils.equals(ConnectorConstant.ORACLE_CONNECTOR, connector)) {
            return;
        }

        // 3、是否首次配置
        OracleMeta meta = super.getMappingMeta(OracleMeta.class);
        if (null == meta) {
            Map<String, String> po = task.getPolicy().get(MappingConstant.POLICY_INCREMENT);
            dictionary = po.get("dictionary");
            meta = new OracleMeta().setScnPos(0L);
            super.saveMappingMeta(meta);
        }

        // 4、获取连接配置
        DatabaseConfig cfg = (DatabaseConfig) mapping.getConfig();
        table = cfg.getTable();
        try {
            // 设置登录用户名以DBA身份 cfg.setUsername(cfg.getUsername() + " AS SYSDBA");
            conn = JDBCUtil.getConnection(cfg.getDriverClassName(), cfg.getUrl(), cfg.getUsername(), cfg.getPassword());
            statement = conn.createStatement();
        } catch (Exception e) {
            logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
        }
        if (null == conn || null == statement) {
            JDBCUtil.close(statement, conn);
            return;
        }

        // 5、轮训扫描oracle字典增量数据
        try {
            for (;;) {
                // 检查线程是否已关闭
                if (!isRunnable) {
                    break;
                }
                pull();
                Thread.sleep(MILLIS);
            }
        } catch (InterruptedException e) {
            logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
        } finally {
            // 释放连接
            JDBCUtil.close(statement, conn);
        }
    }

    private void pull() {
        final OracleMeta meta = super.getMappingMeta(OracleMeta.class);
        try {
            // 如果是首次读取则扫描最新日志点
            Long scnPos = meta.getScnPos();
            if (0 == scnPos) {
                logger.info("Scan last record point.");
                scnPos = (Long) LogmnrUtil.parseMaxScn(statement, dictionary);
            }
            // 抽取增量日志
            JSONObject msg = LogmnrUtil.parse(statement, table, false, dictionary, scnPos);
            // 记录下一次记录点
            Long startPos = msg.getLong("scn");
            // 如果有变化，就更新记录点
            if (scnPos != startPos) {
                meta.setScnPos(startPos);
                super.saveMappingMeta(meta);
                super.sendMsg(msg.getJSONArray("msg"));
            }
        } catch (SQLException e) {
            logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
        } catch (JSONException e) {
            logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
        } catch (Exception e) {
            logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
        }

    }

}
